package org.example.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * rdd的算子，也就是rdd的方法
 *
 *
 */
object Spark17_RDD_Operator_CombineByKey {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)


    /**
      * 附加小练习
      *
      * 希望从这里获得相同key的均值
      *
      * 也就是对于key为a的来说，是(a,3)
      * b 则是 (b,4)
      */
    val rdd2 = context.makeRDD(List(
      ("a",1),("a",2),("b",3),
      ("b",4),("b",5),("a",6)),2)

    //  之前的练习，我们可以自己设定初始值，还有一种算子，可以不用自己设置初始值
    // 但是最为替换，你必须自己定义将数据源的第一个初始数据进行相应的转换
    /**
      * 第一个参数 相同key的第一个数据进行结构的转换
      * 第二个参数 分区内的计算规则
      * 第三个参数 分区间的计算规则
      */
    val newRdd:RDD[(String,(Int,Int))] = rdd2.combineByKey(
        v=>(v,1),//把第一个相同key的的value直接转换为元祖，给下面计算
        //由于这个转换的类型，必须要动态的识别，所以
      // 下面的类型参数是一定要指定的，也就是这里的第一个参数，v的含义4
      (t:(Int,Int),v)=>{
          (t._1+v,t._2+1)
        },
        (t1:(Int,Int),t2:(Int,Int))=>{
          (t1._1+t2._1,t1._2+t2._2)
        }
    )

    context.stop()
  }
}
